package com.cqc.rxjava01.activity;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.view.View;
import android.widget.Toast;

import com.cqc.rxjava01.Http.ApiConstant;
import com.cqc.rxjava01.Http.UserApi;
import com.cqc.rxjava01.R;
import com.cqc.rxjava01.entity.User;
import com.cqc.rxjava01.entity.BaseResult;
import com.cqc.rxjava01.util.LogUtil;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import retrofit2.Retrofit;
import retrofit2.adapter.rxjava2.RxJava2CallAdapterFactory;
import retrofit2.converter.gson.GsonConverterFactory;
import retrofit2.converter.scalars.ScalarsConverterFactory;

/**
 * RxJava基本使用
 * 这可能是最好的RxJava 2.x 入门教程
 * https://www.jianshu.com/p/a93c79e9f689
 */
public class RxJavaOperator01Activity extends AppCompatActivity {

    private CompositeDisposable com;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_operator01);
        com = new CompositeDisposable();
    }

    /**
     rxJava01(View view);
     rxJava02(View view);
     rxJava03(View view);
     rxJava04(View view);
     rxJava05(View view);
     rxJava06(View view);
     rxJava07(View view);
     rxJava08(View view);//注册后登录
     rxJava09(View view);//reJava写法 注册后登录
     rxJava10(View view);//map
     rxJava11(View view);//flatMap
     rxJava12(View view);//concatMap
     */


    /**
     * concatMap：数据类型转换，保证有序
     */
    public void rxJava10(View view) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("String:" + integer);
                }
                return Observable.fromIterable(list).delay(3, TimeUnit.SECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                LogUtil.d(s);
            }
        });
    }

    /**
     * flatMap：数据类型转换，但无法保证有序
     */
    public void rxJava09(View view) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("String：" + integer);
                }
                /**
                 * integer = 1时，被拆成了3个事件
                 * integer = 2时，被拆成了3个事件
                 * integer = 3时，被拆成了3个事件
                 * 总共9个事件，而且是无序的。
                 */

                return Observable.fromIterable(list).delay(5000, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                LogUtil.d(s);
            }
        });
    }

    /**
     * map数据类型转换
     *
     * @param view
     */
    public void rxJava08(View view) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(@NonNull Integer integer) throws Exception {
                return "数字：" + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                LogUtil.d(s);
            }
        });
    }



    /**
     * 链式调用2
     */
    public void rxJava07(View view) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {

            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {

            }
        });
    }

    /**
     * 链式调用1  Observer
     */
    public void rxJava06(View view) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                LogUtil.d("发送," + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Integer integer) {
                LogUtil.d("接收," + Thread.currentThread().getName());
                LogUtil.d("接收, integer=" + integer);

            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                //如果上游不调用emitter.onComplete();则此不会走方法
                LogUtil.d("接收,onComplete  " + Thread.currentThread().getName());
            }
        });
    }

    /**
     * 若上游有多个线程，只有第一个线程有效
     * 若下游有多个线程，只有最后一个线程有效
     */
    public void rxJava05(View view) {
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                LogUtil.d("发送," + Thread.currentThread().getName());
                emitter.onNext(1);
            }
        });

        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                LogUtil.d("observer," + Thread.currentThread().getName());
                LogUtil.d("integer=" + integer);
            }
        };
        //        observable.subscribe(consumer);
        observable
                .subscribeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .observeOn(Schedulers.io())
                .subscribe(consumer);
    }

    /**
     * 转换线程
     */
    public void rxJava04(View view) {
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                LogUtil.d("发送" + Thread.currentThread().getName() + "，1  2");
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onComplete();
            }
        });

        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                LogUtil.d("接收=" + Thread.currentThread().getName() + "，integer=" + integer);
                LogUtil.d("" + integer);
            }
        };

        observable.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(consumer);
    }

    /**
     * onComplete onError 后还可以发送，但下游接收不到
     */
    public void rxJava03(View view) {
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                LogUtil.d("发送" + Thread.currentThread().getName() + "，Hello");
                e.onNext("Hello");
                e.onNext("World");
                //                e.onError(new Throwable("error"));
                e.onComplete();
                e.onNext("1");
            }
        });

        Consumer<String> consumer = new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                LogUtil.d("接收=" + Thread.currentThread().getName() + "，s=" + s);
            }
        };
        observable.subscribe(consumer);
    }


    /**
     * Subscriber  不能作为参数 observable.subscribe(subscriber);
     */
    public void rxJava02(View view) {
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                LogUtil.d("发送" + Thread.currentThread().getName() + "，Hello");
                e.onNext("Hello");
                e.onNext("World");
                //                e.onError(new Throwable("error"));
                e.onComplete();
                e.onNext("1");
            }
        });

        Subscriber<String> subscriber = new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription s) {

            }

            @Override
            public void onNext(String s) {

            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {

            }
        };

        //错误
        //        observable.subscribe(subscriber);
    }

    /**
     * Observer
     */
    public void rxJava01(View view) {

        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                LogUtil.d("发送" + Thread.currentThread().getName() + "，Hello");
                e.onNext("Hello");
                e.onNext("World");
                //                e.onError(new Throwable("error"));
                e.onComplete();//之后仍然可以发送，但是下游不再接收
                e.onNext("1");
            }
        });

        /**
         * Observer 是接口
         */
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                LogUtil.d("onSubscribe");
                com.add(d);
            }

            @Override
            public void onNext(@NonNull String s) {
                LogUtil.d(s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                LogUtil.d("onError");
            }

            @Override
            public void onComplete() {
                LogUtil.d("onComplete");
            }
        };

        observable.subscribe(observer);
    }


    @Override
    protected void onDestroy() {
        super.onDestroy();
        com.clear();
    }
}
